package top.guoziyang.rpc.registry.zk;

import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import top.guoziyang.rpc.entity.RpcRequest;
import top.guoziyang.rpc.enumeration.LoadBalanceEnum;
import top.guoziyang.rpc.enumeration.RpcError;
import top.guoziyang.rpc.exception.RpcException;
import top.guoziyang.rpc.extension.ExtensionLoader;
import top.guoziyang.rpc.loadbalancer.zk.LoadBalance;
import top.guoziyang.rpc.registry.ServiceDiscovery;
import top.guoziyang.rpc.registry.zk.util.CuratorUtils;
import top.guoziyang.rpc.util.CollectionUtil;

import java.net.InetSocketAddress;
import java.util.List;

/**
 * service discovery based on zookeeper
 *
 * @author shuang.kou
 * @createTime 2020年06月01日 15:16:00
 */
@Slf4j
public class ZkServiceDiscoveryImpl implements ServiceDiscovery {
    private final LoadBalance loadBalance;

    public ZkServiceDiscoveryImpl() {//换成了此rpc的轮转算法
        this.loadBalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(LoadBalanceEnum.LOADBALANCE.getName());
    }

    @Override
    public InetSocketAddress lookupService(RpcRequest rpcRequest) {
        //获取RPC请求的服务名称
        String rpcServiceName = rpcRequest.getRpcServiceName();
        //返回一个CuratorFramework类型的ZooKeeper客户端对象zkClient，用于连接和与ZooKeeper服务器进行交互
        CuratorFramework zkClient = CuratorUtils.getZkClient();
        //这个方法会查询ZooKeeper服务器上指定节点的子节点列表，这些子节点一般用于存储服务的地址信息。
        List<String> serviceUrlList = CuratorUtils.getChildrenNodes(zkClient, rpcServiceName);
        if (CollectionUtil.isEmpty(serviceUrlList)) {
            //换一个同等的报错枚举类
            throw new RpcException(RpcError.SERVICE_NOT_FOUND, rpcServiceName);
        }
        // load balancing
        //轮转算法获取一个服务地址
        String targetServiceUrl = loadBalance.selectServiceAddress(serviceUrlList, rpcRequest);
        log.info("Successfully found the service address:[{}]", targetServiceUrl);
        String[] socketAddressArray = targetServiceUrl.split(":");
        String host = socketAddressArray[0];
        int port = Integer.parseInt(socketAddressArray[1]);
        return new InetSocketAddress(host, port);
    }
}
